/**
 * File: ResultMergeProcessor.java
 * Created by: mhaimel
 * Created on: 23 Jul 2009
 * CVS:  $Id: ResultMergeProcessor.java,v 1.3 2009/12/07 10:03:38 mhaimel Exp $
 */
package uk.ac.ebi.curtain.processor.velvet.impl;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.PrintStream;
import java.io.PrintWriter;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import uk.ac.ebi.curtain.controller.velvet.VelvetOutputBean;
import uk.ac.ebi.curtain.processor.AbstractQueueProcessor;
import uk.ac.ebi.curtain.util.FileHelper;
import uk.ac.ebi.curtain.utils.concurrent.BlockingQueueShutdown;
import uk.ac.ebi.curtain.utils.data.FileType;
import uk.ac.ebi.curtain.utils.file.EntryMapper;
import uk.ac.ebi.curtain.utils.file.EntryMapperBean;
import uk.ac.ebi.curtain.utils.file.WriteUtil;
import uk.ac.ebi.curtain.utils.io.impl.FileIO;

/**
 * @author mhaimel
 *
 */
public class ResultMergeProcessor extends AbstractQueueProcessor<VelvetOutputBean, VelvetOutputBean> {
	private static final Integer MIN_OUTPUT_LENGTH = 60;
	private final AtomicLong contigCnt = new AtomicLong(0);
	
	private final AtomicReference<BlockingQueueShutdown<ByteBuffer>> output = new AtomicReference<BlockingQueueShutdown<ByteBuffer>>();
	private final AtomicInteger minOutputLength = new AtomicInteger();
	private final FileHelper fileHelper = new FileHelper();
	
	public ResultMergeProcessor() {
		this(MIN_OUTPUT_LENGTH);
	}

	public ResultMergeProcessor(Integer minOutLen) {
		this.minOutputLength.set(minOutLen);
	}
	
	public void setOutput(BlockingQueueShutdown<ByteBuffer> output) {
		this.output.set(output);
	}

	public BlockingQueueShutdown<ByteBuffer> getOutput() {
		return output.get();
	}

	@Override
	protected VelvetOutputBean buildReturnObject() {
		return null;
	}
	
	@Override
	protected VelvetOutputBean doProcess(VelvetOutputBean input) throws InterruptedException {
		getLog().debug("Collect results from " + input.getOutputContig() +" ..." );
		FileType fType = FileType.fasta;
		
		ByteArrayOutputStream out = new ByteArrayOutputStream();
		PrintWriter pw = new PrintWriter(new PrintStream(out));
		MyMapper mapper = new MyMapper(fType.getWriter(), pw , getMinOutputLength(), true);
		input.getOutputContig().parse(mapper);
		pw.flush();
		FileIO.closeQuietly(pw);
		ByteBuffer buffer = ByteBuffer.wrap(out.toByteArray());
		getOutput().put(buffer);
		getLog().debug("Finished collecting results from " + input.getOutputContig() +"!" );
		return input;
	}


	private Integer getMinOutputLength(){
		return minOutputLength.get(); 
	}
	
	public void setMinOutputLength(Integer minLength) {
		this.minOutputLength.set(minLength);
	}

	private boolean isValid(EntryMapperBean bean, Integer minLen) {
		return bean.getSequence().length() >= minLen;
	}
	
	private EntryMapperBean rebuildId(EntryMapperBean bean){
		EntryMapperBean nBean = new EntryMapperBean(bean);
		long nId = contigCnt.getAndIncrement();
		int sequLen = bean.getSequenceLength();
		File file = bean.getInfo().getFile().getParentFile(); // get test dir
		if(!fileHelper.getNSize().equals(file.getName().length())){
			file = file.getParentFile(); // remove "test" dir name
		}
		Integer dirId = fileHelper.idFromSubDirectory(file);
		nBean.setId("NODE_"+nId+"_length_"+sequLen+"_bucket_"+dirId);
		return nBean;
	}

	private class MyMapper implements EntryMapper{
		
		private final PrintWriter out;
		private final WriteUtil writer;
		private final Integer minLen;
		private final boolean rewriteId;

		public MyMapper(WriteUtil writer, PrintWriter out, Integer minLen, boolean rewriteId) {
			this.writer = writer;
			this.out = out;
			this.minLen = minLen;
			this.rewriteId = rewriteId;
		}

		public void mapEntry(EntryMapperBean bean) {
			if(isValid(bean, minLen)){
				if(rewriteId){
					bean = rebuildId(bean);
				}
				writer.write(out, bean);
			}
		}
	}


}
